package cn.monkey.io.transport.netty;

import cn.monkey.io.transport.core.Connection;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.AttributeKey;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.function.Predicate;

public class DefaultChannelConnection implements ChannelConnection {

    protected final Channel channel;

    protected final ChannelFluxReceive receiver;

    public DefaultChannelConnection(Channel channel) {
        this.channel = channel;
        this.receiver = new ChannelFluxReceive(this);
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver.filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }

    @Override
    public boolean isActive() {
        return this.channel.isActive();
    }

    @Override
    public Mono<Void> close() {
        this.receiver.dispose();
        ChannelFuture channelFuture = this.channel.close();
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (!future.isSuccess()) {
                sink.error(future.cause());
            }
            sink.success();
        }));
    }

    @Override
    public ChannelConnection bind() {
        this.channel.attr(AttributeKey.valueOf(Connection.KEY)).set(this);
        return this;
    }

    @Override
    public Mono<Void> write(Object o) {
        ChannelFuture channelFuture = this.channel.write(o);
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(future.cause());
        }));
    }

    @Override
    public Mono<Void> flush() {
        this.channel.flush();
        return Mono.empty();
    }

    @Override
    public Mono<Void> writeAndFlush(Object o) {
        ChannelFuture channelFuture = this.channel.writeAndFlush(o);
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(future.cause());
        }));
    }

    @Override
    public Channel channel() {
        return this.channel;
    }
}
